package com.afdor.rws.event.driven.channel.aliyun.mns.consumer.topic;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.driven.api.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.aliyun.mns.config.AliyunMnsEventConfig;
import com.afdor.rws.event.driven.channel.aliyun.mns.config.AliyunMnsEventConsumerConfig;
import com.afdor.rws.event.driven.channel.aliyun.mns.utils.AliyunMnsEventHolder;
import com.afdor.rws.event.driven.channel.aliyun.mns.utils.AliyunMnsUtils;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.support.channel.consumer.AbstractRunnableEventConsumer;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.CloudTopic;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import com.aliyun.mns.model.SubscriptionMeta;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 阿里云MNS(消息服务)基于topic(主题)实现的消费者
 *
 * @author 悭梵
 * @date Created in 2018-12-13 15:06
 */
@Slf4j
@NoArgsConstructor
@Extension(value = "aliyunMnsTopic", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class AliyunMnsEventConsumer extends AbstractRunnableEventConsumer<AliyunMnsEventHolder, AliyunMnsEventConsumerConfig> {
    @Getter
    private CloudQueue topicSubQueue;
    @Getter
    private CloudAccount cloudAccount;

    @Override
    protected void doStart() {
        if (this.cloudAccount != null) {
            throw new RuntimeException("已执行初始化操作，请不要再重复执行初始化操作了。");
        }

        AliyunMnsEventConfig aliyunConfig = getConfig().getEventConfig();
        this.cloudAccount = new CloudAccount(aliyunConfig.getSecretId(), aliyunConfig.getSecretKey(), aliyunConfig.getEndpoint());

        String topicName = AliyunMnsUtils.getTopicName(getEventSourceName());
        String topicSubQueueName = AliyunMnsUtils.getTopicSubQueueName(getEventSourceName(), getConfig().getConsumerName());

        MNSClient mnsClient = getCloudAccount().getMNSClient();

        /* 获取子队列 */
        this.topicSubQueue = mnsClient.getQueueRef(topicSubQueueName);
        if (!this.topicSubQueue.isQueueExist()) {
            this.topicSubQueue.create();
        }

        CloudTopic topic = mnsClient.getTopicRef(topicName);
        topic.create();

        /* 创建订阅关系关系 */
        SubscriptionMeta meta = new SubscriptionMeta();
        /* 主题名称 */
        meta.setTopicName(topicName);
        /* 路由队列名称 */
        meta.setSubscriptionName(topicSubQueueName);
        /* 消息格式 */
        meta.setNotifyContentFormat(SubscriptionMeta.NotifyContentFormat.SIMPLIFIED);
        /* 接收消息的终端地址 */
        meta.setEndpoint(topic.generateQueueEndpoint(topicSubQueueName));
        topic.subscribe(meta);
    }


    @Override
    public EventRetryStrategy getEventRetryStrategy() {
        EventRetryStrategy eventRetryStrategy = super.getEventRetryStrategy();
        if (eventRetryStrategy == null) {
            return ExtensionLoader.getInstance(EventRetryStrategy.class)
                    .newExtension("aliyunMns", this);
        }
        return eventRetryStrategy;
    }

    @Override
    protected AliyunMnsEventHolder getEvent() {
        EventConsumerConfig eventConsumerConfig = getConfig();
        int waitSeconds = 0;
        if (eventConsumerConfig instanceof AliyunMnsEventConsumerConfig) {
            waitSeconds = ((AliyunMnsEventConsumerConfig) eventConsumerConfig).getWaitSeconds();
        }
        Message message = waitSeconds > 0 ? getTopicSubQueue().popMessage(waitSeconds) : getTopicSubQueue().popMessage();
        if (message == null) {
            return null;
        }
        return new AliyunMnsEventHolder(getSerialization().unserialize(message.getMessageBodyAsRawString()), message.getDequeueCount(), message.getReceiptHandle());
    }

    @Override
    protected List<AliyunMnsEventHolder> getEvents(int batchSize) {
        EventConsumerConfig eventConsumerConfig = getConfig();
        int waitSeconds = 0;
        if (eventConsumerConfig instanceof AliyunMnsEventConsumerConfig) {
            waitSeconds = ((AliyunMnsEventConsumerConfig) eventConsumerConfig).getWaitSeconds();
        }
        if (batchSize <= 0) {
            batchSize = 50;
        }
        List<Message> messageList = waitSeconds > 0 ? this.getTopicSubQueue().batchPopMessage(batchSize, waitSeconds) : getTopicSubQueue().batchPopMessage(batchSize);
        if (messageList == null || messageList.isEmpty()) {
            return null;
        }
        return messageList.parallelStream()
                .map(message -> new AliyunMnsEventHolder(getSerialization().unserialize(message.getMessageBodyAsRawString()), message.getDequeueCount(), message.getReceiptHandle()))
                .collect(Collectors.toList());
    }

    @Override
    protected void doSuccess(AliyunMnsEventHolder event, Object result) {
        getTopicSubQueue().deleteMessage(event.getReceiptHandle());
    }

    @Override
    protected void doFailure(AliyunMnsEventHolder event, String message) {
        getTopicSubQueue().deleteMessage(event.getReceiptHandle());
    }

    @Override
    protected void doStop() {
        super.doStop();
        if (this.topicSubQueue != null) {
            this.topicSubQueue = null;
        }
        if (this.cloudAccount != null) {
            this.cloudAccount.getMNSClient().close();
            this.cloudAccount = null;
        }
    }
}
